package com.jumei.flume.sink.kudu;

import com.alibaba.fastjson.JSON;
import com.jumei.bi.kudu.ReloadableKuduClient;
import com.jumei.flume.common.StringUtils;
import com.jumei.flume.annotation.Column;
import com.jumei.flume.annotation.DateAddField;
import com.jumei.flume.annotation.Table;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.AnnotationUtils;

import java.lang.reflect.Field;
import java.util.*;

/**
 * Created by ZhongGang
 * at 2017/7/5 15:59
 */
public class KuduJdbcDaoImpl implements KuduJdbcDao {

    private static final Logger LOGGER = LoggerFactory.getLogger(KuduJdbcDaoImpl.class);

    private ReloadableKuduClient kuduClient;

    public KuduJdbcDaoImpl(ReloadableKuduClient kuduClient) {
		super();
		this.kuduClient = kuduClient;
	}

	@Override
    public int upsert(String table, List<Map<String, Object>> records) throws Exception {
        if (records == null || records.isEmpty()) {
            return 0;
        }

        int count = 0;

        KuduTable kuduTable;
        KuduSession kuduSession = null;

        try {
            kuduTable = kuduClient.openTable(table);
            Schema schema = kuduTable.getSchema();
            kuduSession = kuduClient.newSession();
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
            for (int i = 1; i <= records.size(); i++) {
                Map<String, Object> record = records.get(i - 1);
                if (record.isEmpty()) {
                    continue;
                }
                Upsert upsert = kuduTable.newUpsert();
                PartialRow row = upsert.getRow();
                Set<Map.Entry<String, Object>> entries = record.entrySet();
                for (Map.Entry<String, Object> entry : entries) {
                    String columnName = entry.getKey().toLowerCase();
                    Object columnValue = entry.getValue();
                    if (columnValue == null) {
                        row.setNull(columnName);
                    } else {
                        ColumnSchema columnSchema = schema.getColumn(columnName);
                        Type type = columnSchema.getType();
                        String value = columnValue.toString();
                        switch (type) {
                            case INT8:
                                row.addByte(columnName, Byte.valueOf(value));
                                break;
                            case INT16:
                                row.addShort(columnName, Short.valueOf(value));
                                break;
                            case INT32:
                                row.addInt(columnName, Integer.valueOf(value));
                                break;
                            case INT64:
                                row.addLong(columnName, Long.valueOf(value));
                                break;
                            case STRING:
                                row.addString(columnName, value);
                                break;
                            case BOOL:
                                row.addBoolean(columnName, Boolean.valueOf(value));
                                break;
                            case FLOAT:
                                row.addFloat(columnName, Float.valueOf(value));
                                break;
                            case DOUBLE:
                                row.addDouble(columnName, Double.valueOf(value));
                                break;
                            case BINARY:
                                throw new UnsupportedOperationException(String.format("column name is %s, column value is %s, column type is %s", columnName, columnValue, columnValue.getClass().getName()));
                            default:
                                throw new UnsupportedOperationException(String.format("column name is %s, column value is %s, column type is %s", columnName, columnValue, columnValue.getClass().getName()));
                        }
                    }
                }
                kuduSession.apply(upsert);
                if (i % BATCH_FLUSH_SIZE == 0) {
                    List<OperationResponse> operationResponses = kuduSession.flush();
                    for (OperationResponse operationResponse : operationResponses) {
                        if (operationResponse.hasRowError()) {
                            LOGGER.error("it is failure to upsert the record into kudu table {}, the error is {}, original message is {}.", 
                            		table, operationResponse.getRowError().toString(), JSON.toJSONString(record));
                            
                            
                            //TODO：不阻塞主流程
                            //throw new RuntimeException("kudu upsert error : " + operationResponse.getRowError().toString());
                        } else {
                            count += 1;
                        }
                    }
                }
            }

            List<OperationResponse> operationResponses = kuduSession.flush();
            for (OperationResponse operationResponse : operationResponses) {
                if (operationResponse.hasRowError()) {
                    LOGGER.error("it is failure to upsert the record into kudu table {}, the error is {}", table, operationResponse.getRowError().toString());
                    
                  //TODO：不阻塞主流程
                    //throw new RuntimeException("kudu upsert error : " + operationResponse.getRowError().toString());
                } else {
                    count += 1;
                }
            }

        } catch (Exception e) {
            LOGGER.error("it is failure to upsert the record(s) into kudu table {}.", table, e);
            throw e;
        } finally {
            try {
                if (kuduSession != null) {
                    kuduSession.close();
                }
            } catch (Exception e) {
                LOGGER.error("it is failure to close the kudu session.", e);
            }
        }

        return count;
    }

    @Override
    public boolean upsert(String table, Map<String, Object> record) throws Exception {
        List<Map<String, Object>> records = new ArrayList<>();
        records.add(record);
        return upsert(table, records) == 1;
    }

    @Override
    public <T> int upsert(List<T> records) throws Exception {
        if (records == null || records.isEmpty()) {
            return 0;
        }

        Table table = AnnotationUtils.findAnnotation(records.get(0).getClass(), Table.class);
        String tableName = table.name();
        String addTimeField = table.addTimeField();

        List<Map<String, Object>> results = new ArrayList<>();
        for (T record : records) {
            if (record != null) {
                Map<String, Object> result = new HashMap<>();
                Field[] fields = record.getClass().getDeclaredFields();
                for (Field field : fields) {
                    field.setAccessible(true);
                    Object fieldValue = field.get(record);
                    if (fieldValue != null) {
                        String filedName = StringUtils.camelToUnderline(field.getName());
                        Column column = AnnotationUtils.findAnnotation(field, Column.class);
                        
//                        if(column.isIgnore()) {//是否持久化
//                        	continue;
//                        }
                        
                        if (!StringUtils.isEmptyStringTrim(column.name())) {
                            filedName = column.name();
                        }

                        //日期处理, 日期类型统一转换成时间戳
                        if (fieldValue instanceof Date) {
                            long time = ((Date) fieldValue).getTime();

                            //处理日期增加注解器
                            DateAddField dateAddField = AnnotationUtils.findAnnotation(field, DateAddField.class);
                            if (dateAddField != null && !StringUtils.isEmptyStringTrim(dateAddField.fieldName())) {
                                DateTime dateTime = new DateTime(time);
                                String addFieldValue = dateTime.toString("yyyyMMdd");
                                result.put(dateAddField.fieldName(), Long.valueOf(addFieldValue));
                            }

                            fieldValue = time;
                        }

                        if (fieldValue instanceof String) {
                            fieldValue = fieldValue.toString().replace("\r", "").replace("\n", "").replace("\r\n", "");
                        }

                        result.put(filedName, fieldValue);
                    }
                }

                if (!StringUtils.isEmptyStringTrim(addTimeField)) {
                    result.put(addTimeField, System.currentTimeMillis());
                }
                results.add(result);
            }
        }

        return upsert(tableName, results);
    }

    @Override
    public <T> boolean upsert(T record) throws Exception {
        List<T> records = new ArrayList<>();
        records.add(record);
        return upsert(records) == 1;
    }

    @Override
    public <T> int delete(List<T> records) throws Exception {
        if (records == null || records.isEmpty()) {
            return 0;
        }

        Table table = AnnotationUtils.findAnnotation(records.get(0).getClass(), Table.class);
        String tableName = table.name();
        String addTimeField = table.addTimeField();

        List<Map<String, Object>> results = new ArrayList<>();
        for (T record : records) {
            if (record != null) {
                Map<String, Object> result = new HashMap<>();
                Field[] fields = record.getClass().getDeclaredFields();
                for (Field field : fields) {
                    field.setAccessible(true);
                    Object fieldValue = field.get(record);
                    if (fieldValue != null) {
                        String filedName = StringUtils.camelToUnderline(field.getName());
                        Column column = AnnotationUtils.findAnnotation(field, Column.class);
                        if (!StringUtils.isEmptyStringTrim(column.name())) {
                            filedName = column.name();
                        }

                        result.put(filedName, fieldValue);
                    }
                }

                if (!StringUtils.isEmptyStringTrim(addTimeField)) {
                    result.put(addTimeField, System.currentTimeMillis());
                }
                results.add(result);
            }
        }

        return delete(tableName, results);
    }

    @Override
    public <T> boolean delete(T record) throws Exception {
        List<T> records = new ArrayList<>();
        records.add(record);
        return delete(records) == 1;
    }

    @Override
    public int delete(String table, List<Map<String, Object>> records) throws Exception {

        if (records == null || records.isEmpty()) {
            return 0;
        }

        KuduTable kuduTable;
        KuduSession kuduSession = null;

        int count = 0;
        try {
            kuduTable = kuduClient.openTable(table);
            Schema schema = kuduTable.getSchema();
            kuduSession = kuduClient.newSession();
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
            for (int i = 1; i <= records.size(); i++) {
                Map<String, Object> record = records.get(i - 1);
                Delete delete = kuduTable.newDelete();
                PartialRow row = delete.getRow();
                Set<Map.Entry<String, Object>> entries = record.entrySet();
                for (Map.Entry<String, Object> entry : entries) {
                    String columnName = entry.getKey();
                    Object columnValue = entry.getValue();
                    if (columnValue == null) {
                        row.setNull(columnName);
                    } else {
                        ColumnSchema columnSchema = schema.getColumn(columnName);
                        Type type = columnSchema.getType();
                        switch (type) {
                            case INT8:
                                row.addByte(columnName, Byte.valueOf(String.valueOf(columnValue)));
                                break;
                            case INT16:
                                row.addShort(columnName, Short.valueOf(String.valueOf(columnValue)));
                                break;
                            case INT32:
                                row.addInt(columnName, Integer.valueOf(String.valueOf(columnValue)));
                                break;
                            case INT64:
                                row.addLong(columnName, Long.valueOf(String.valueOf(columnValue)));
                                break;
                            case BINARY:
                                throw new UnsupportedOperationException(String.format("column name is %s, column value is %s, column type is %s", columnName, columnValue, columnValue.getClass().getName()));
                            case STRING:
                                row.addString(columnName, String.valueOf(columnValue));
                                break;
                            case BOOL:
                                row.addBoolean(columnName, Boolean.valueOf(String.valueOf(columnValue)));
                                break;
                            case FLOAT:
                                row.addFloat(columnName, Float.valueOf(String.valueOf(columnValue)));
                                break;
                            case DOUBLE:
                                row.addDouble(columnName, Double.valueOf(String.valueOf(columnValue)));
                                break;
                            default:
                                throw new UnsupportedOperationException(String.format("column name is %s, column value is %s, column type is %s", columnName, columnValue, columnValue.getClass().getName()));
                        }
                    }
                }
                kuduSession.apply(delete);
                if (i % BATCH_FLUSH_SIZE == 0) {
                    List<OperationResponse> operationResponses = kuduSession.flush();
                    for (OperationResponse operationResponse : operationResponses) {
                        if (operationResponse.hasRowError()) {
                            LOGGER.error("it is failure to delete the record from kudu table {}, the error is {}", table, operationResponse.getRowError().toString());
                            if (operationResponse.getRowError().toString().contains("status=Not found: key not found (error 0)")) {
                                continue;
                            }
                            throw new RuntimeException("kudu delete error : " + operationResponse.getRowError().toString());
                        } else {
                            count += 1;
                        }
                    }
                }
            }

            List<OperationResponse> operationResponses = kuduSession.flush();
            for (OperationResponse operationResponse : operationResponses) {
                if (operationResponse.hasRowError()) {
                    LOGGER.error("it is failure to delete the record from kudu table {}, the error is {}", table, operationResponse.getRowError().toString());
                    if (operationResponse.getRowError().toString().contains("status=Not found: key not found (error 0)")) {
                        continue;
                    }
                    throw new RuntimeException("kudu delete error : " + operationResponse.getRowError().toString());
                } else {
                    count += 1;
                }
            }
        } catch (Exception e) {
            LOGGER.error("it is failure to delete the record from kudu.", e);
            throw e;
        } finally {
            try {
                if (kuduSession != null) {
                    kuduSession.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return count;
    }

    @Override
    public int delete(String table, Map<String, Object> record) throws Exception {
        List<Map<String, Object>> records = new ArrayList<>();
        records.add(record);
        return delete(table, records);
    }

	public ReloadableKuduClient getKuduClient() {
		return kuduClient;
	}

	public void setKuduClient(ReloadableKuduClient kuduClient) {
		this.kuduClient = kuduClient;
	}
}
